package com.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

/**
 * Created by IntelliJ IDEA.
 * <p>
 * Author: libingyang
 * Date: 2021-02-22 11:50
 */
public class CustomOffsetConsumer {


    public static Map currentOffset = new HashMap();


    public static void main(String[] args) {

        Properties props = new Properties();
        // 定义 kakfa 服务的地址，不需要将所有 broker 指定上
        props.put("bootstrap.servers", "192.168.100.125:9092");

        // 制定 consumer group
        props.put("group.id", "test");

//        // 是否自动确认 offset
//        props.put("enable.auto.commit", "true");

        // 自动确认 offset 的时间间隔
        props.put("auto.commit.interval.ms", "1000");

        // key 的序列化类
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        // value 的序列化类
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        // 定义 consumer
        final KafkaConsumer<String, String> consumer = new KafkaConsumer(props);

        // 消费者订阅的 topic, 可同时订阅多个
        consumer.subscribe(Arrays.asList("first", "second", "third"), new ConsumerRebalanceListener() {

            // 会在rebalance操作之前调用，用于我们提交消费者偏移
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {

                System.out.println("*- in ralance:onPartitionsRevoked");

                commitOffset(currentOffset);

            }


            // OnPartitionAssigned会在rebalance操作之后调用，用于我们拉取新的分配区的偏移量
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {

                currentOffset.clear();

                for (TopicPartition partition : partitions) {

                    consumer.seek(partition, getOffset(partition));//定位到最近提交的offset位置继续消费
                }
            }
        });

        while (true) {
            // 读取数据，读取超时时间为 100ms
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {

                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

                currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
            }

            commitOffset(currentOffset);//异步提交
        }
    }

    /**
     * 自定义实现获取offset
     *
     * @param partition
     * @return
     */
    private static long getOffset(TopicPartition partition) {

        return 0;
    }

    /**
     * 自定义实现保存offset
     */
    private static void commitOffset(Map currentOffset) {

    }

}
